package com.fudan.es.service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.nested.ReverseNested;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.metrics.Avg;
import org.elasticsearch.search.aggregations.metrics.Sum;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.fudan.biz.data.entity.BigDataSource;
import com.fudan.biz.data.service.BigDataSourceService;
import com.fudan.biz.data.service.FieldMetaService;
import com.fudan.es.Aggs;
import com.fudan.es.EsClient;
import com.fudan.es.aggs.AggsBuilder;
import com.fudan.es.entity.FieldMeta;
import com.fudan.es.entity.Metric;
import com.fudan.es.pack.DateAggsPack;
import com.fudan.es.pack.TermAggsPack;

@Service
public class AggsService {
	@Autowired
	private FieldMetaService fieldMetaService;
	@Autowired
	private EsClient esClient;
	@Autowired
	private BigDataSourceService bigDataSourceService;

	static class AggsPair {
		private String field;
		private Object val;

		public String getField() {
			return field;
		}

		public void setField(String field) {
			this.field = field;
		}

		public Object getVal() {
			return val;
		}

		public void setVal(Object val) {
			this.val = val;
		}

		static AggsPair build(String field, Object val) {
			AggsPair aggsPair = new AggsPair();
			aggsPair.setField(field);
			aggsPair.setVal(val);
			return aggsPair;
		}
	}

	public List<Map<String, Object>> aggs(String dsId, List<Metric> metrics, String dimension,int dimensionSize, QueryBuilder query)
			throws Exception {
		List<Map<String, Object>> termResponse = new ArrayList<>();
		SearchSourceBuilder aggregation = new SearchSourceBuilder().size(0);
		if (query != null) {
			aggregation.query(query);
		}
		String rex = "([^(]+)[(]([0-9]*)[)]";
		if (dimension != null && !"".equals(dimension)) {
			Pattern pattern = Pattern.compile(rex);
			Matcher matcher = pattern.matcher(dimension);
			if (matcher.find()) {
				String fuc = matcher.group(1);
				String dimensionId = matcher.group(2);
				if ("terms".equalsIgnoreCase(fuc)) {
					FieldMeta terms = fieldMetaService.get(dimensionId);
					TermAggsPack rootPack = new TermAggsPack(terms,dimensionSize);
					if (metrics != null) {
						for (Metric metric : metrics) {
							// Metric metric = Metric.build(Metrics.COUNT, field);
							rootPack.addMetric(metric);
						}
					}
					aggregation.aggregation(rootPack.getRootAggs());
					System.out.println(aggregation);
					BigDataSource dataSource = bigDataSourceService.get(dsId);
					RestHighLevelClient client = esClient.getClient(dataSource.getDwId());
					SearchResponse response = client.search(new SearchRequest(dataSource.getPath()).source(aggregation),
							RequestOptions.DEFAULT);
					termResponse = rootPack.getTermResponse(response);
				}
				if("date".equalsIgnoreCase(fuc)) {
					FieldMeta field = fieldMetaService.get(dimensionId);
					DateAggsPack rootPack = new DateAggsPack(field, DateHistogramInterval.MONTH);
					if (metrics != null) {
						for (Metric metric : metrics) {
							// Metric metric = Metric.build(Metrics.COUNT, field);
							rootPack.addMetric(metric);
						}
					}
					aggregation.aggregation(rootPack.getRootAggs());
					System.out.println(aggregation);
					BigDataSource dataSource = bigDataSourceService.get(dsId);
					RestHighLevelClient client = esClient.getClient(dataSource.getDwId());
					SearchResponse response = client.search(new SearchRequest(dataSource.getPath()).source(aggregation),
							RequestOptions.DEFAULT);
					termResponse = rootPack.getDateResponse(response);
				}
			}
		}
		return termResponse;
	}

	public String getRequest(String dsId, List<Metric> metrics, String dimension,int dimensionSize,QueryBuilder query) {
		SearchSourceBuilder aggregation = new SearchSourceBuilder().size(0);
		if (query != null) {
			aggregation.query(query);
		}
		String rex = "([^(]+)[(]([0-9]*)[)]";
		if (dimension != null && !"".equals(dimension)) {
			Pattern pattern = Pattern.compile(rex);
			Matcher matcher = pattern.matcher(dimension);
			if (matcher.find()) {
				String fuc = matcher.group(1);
				String dimensionId = matcher.group(2);
				if ("terms".equalsIgnoreCase(fuc)) {
					FieldMeta terms = fieldMetaService.get(dimensionId);
					TermAggsPack rootPack = new TermAggsPack(terms,dimensionSize);
					if (metrics != null) {
						for (Metric metric : metrics) {
							rootPack.addMetric(metric);
						}
					}
					aggregation.aggregation(rootPack.getRootAggs());
				}
				if("date".equalsIgnoreCase(fuc)) {
					FieldMeta field = fieldMetaService.get(dimensionId);
					DateAggsPack rootPack = new DateAggsPack(field, DateHistogramInterval.MONTH);
					if (metrics != null) {
						for (Metric metric : metrics) {
							// Metric metric = Metric.build(Metrics.COUNT, field);
							rootPack.addMetric(metric);
						}
					}
					aggregation.aggregation(rootPack.getRootAggs());
				}
			}
		}
		return aggregation.toString();
	}

	public Map<String, Object> countByFields(String dsId, String[] dimensions) throws Exception {
		int level = 0;
		Map<String, Object> ret = new HashMap<>();
		BigDataSource dataSource = bigDataSourceService.get(dsId);
		SearchSourceBuilder aggregation = new SearchSourceBuilder().size(0);
		List<AggsBuilder> countList = new ArrayList<>();
		List<List<String>> termLevelList = new ArrayList<>();
		List<Integer> termLevelL = new ArrayList<>();
		for (String dimension : dimensions) {
			AggsBuilder aggs = null;
			if (dimension != null && !"".equals(dimension)) {
				FieldMeta target = fieldMetaService.get(dimension);
				List<String> termLevel = new ArrayList<>();
				if (target.getInNested()) {
					String path = target.getNestedPath();
					AggregationBuilder termsAggs = Aggs.termsWithMissing(target.getName(), target.getName());
					String[] pattern = path.split("[.]");
					termsAggs.subAggregation(AggregationBuilders.reverseNested("r"));
					level = pattern.length;
					termLevelL.add(level);
					for (int i = 0; i < level; i++) {
						String rp = pattern[0];
						for (int j = 1; j <= i; j++) {
							rp = rp + "." + pattern[j];
						}
						termLevel.add(rp);
						termsAggs.subAggregation(AggregationBuilders.reverseNested("r" + i).path(rp));
					}
					termLevelList.add(termLevel);
					aggs = AggsBuilder.build(Aggs.nested(path, path)).subAggs(AggsBuilder.build(termsAggs));
				} else {
					aggs = AggsBuilder.build(Aggs.termsWithMissing(target.getName(), target.getName()));
				}
				countList.add(aggs);
				aggregation.aggregation(aggs.getAggregationBuilder());
			}
		}
		RestHighLevelClient client = esClient.getClient(dataSource.getDwId());
		SearchRequest r = new SearchRequest(dataSource.getPath()).source(aggregation);
		SearchResponse response = client.search(r, RequestOptions.DEFAULT);
		Map<String, Object> childRet = new HashMap<>();
		for (int p = 0; p < dimensions.length; p++) {
			ret.put(dimensions[p], childRet);
			Map<String, Object> cr = (Map<String, Object>) ret.get(dimensions[p]);
			Terms t = countList.get(p).getAggsList().getAggsResponse(response);
			List<? extends Bucket> buckets = t.getBuckets();
			List<AggsPair> n = new ArrayList<>();
			cr.put("default", n);
			for (Bucket bucket : buckets) {
				((List<AggsPair>) cr.get("default")).add(AggsPair.build(bucket.getKeyAsString(), bucket.getDocCount()));
				if (level != 0) {
					if (!cr.containsKey("root")) {
						List<AggsPair> rm = new ArrayList<>();
						cr.put("root", rm);
					}
					ReverseNested rn = bucket.getAggregations().get("r");
					((List<AggsPair>) cr.get("root")).add(AggsPair.build(bucket.getKeyAsString(), rn.getDocCount()));
					for (int i = 0; i < termLevelL.get(p); i++) {
						String key = termLevelList.get(p).get(i);
						if (!cr.containsKey(key)) {
							List<AggsPair> rmm = new ArrayList<>();
							cr.put(key, rmm);
						}
						ReverseNested rnn = bucket.getAggregations().get("r" + i);
						((List<AggsPair>) cr.get(key)).add(AggsPair.build(bucket.getKeyAsString(), rnn.getDocCount()));
					}
				}
			}
		}
		return ret;
	}

	@SuppressWarnings("unchecked")
	public Map<String, Object> countByField(String dsId, String dimension) throws Exception {
		int level = 0;
		Map<String, Object> ret = new HashMap<>();
		BigDataSource dataSource = bigDataSourceService.get(dsId);
		if (dimension != null && !"".equals(dimension)) {
			FieldMeta target = fieldMetaService.get(dimension);
			List<String> termLevel = new ArrayList<>();
			AggsBuilder aggs = null;
			if (target.getInNested()) {
				String path = target.getNestedPath();
				AggregationBuilder termsAggs = Aggs.termsWithMissing(target.getName(), target.getName());
				String[] pattern = path.split("[.]");
				termsAggs.subAggregation(AggregationBuilders.reverseNested("r"));
				level = pattern.length;
				for (int i = 0; i < level; i++) {
					String rp = pattern[0];
					for (int j = 1; j <= i; j++) {
						rp = rp + "." + pattern[j];
					}
					termLevel.add(rp);
					termsAggs.subAggregation(AggregationBuilders.reverseNested("r" + i).path(rp));
				}

				aggs = AggsBuilder.build(Aggs.nested(path, path)).subAggs(AggsBuilder.build(termsAggs));
			} else {
				aggs = AggsBuilder.build(Aggs.termsWithMissing(target.getName(), target.getName()));
			}
			if (aggs != null) {
				RestHighLevelClient client = esClient.getClient(dataSource.getDwId());
				SearchSourceBuilder aggregation = new SearchSourceBuilder().size(0)
						.aggregation(aggs.getAggregationBuilder());
				SearchRequest r = new SearchRequest(dataSource.getPath()).source(aggregation);
				SearchResponse response = client.search(r, RequestOptions.DEFAULT);
				Terms t = aggs.getAggsList().getAggsResponse(response);
				List<? extends Bucket> buckets = t.getBuckets();
				Map<String, Object> n = new HashMap<>();
				ret.put("default", n);
				for (Bucket bucket : buckets) {
					((Map<String, Object>) ret.get("default")).put(bucket.getKeyAsString(), bucket.getDocCount());
					if (level != 0) {
						if (!ret.containsKey("root")) {
							Map<String, Object> rm = new HashMap<>();
							ret.put("root", rm);
						}
						ReverseNested rn = bucket.getAggregations().get("r");
						((Map<String, Object>) ret.get("root")).put(bucket.getKeyAsString(), rn.getDocCount());
						for (int i = 0; i < level; i++) {
							String key = termLevel.get(i);
							if (!ret.containsKey(key)) {
								Map<String, Object> rmm = new HashMap<>();
								ret.put(key, rmm);
							}
							ReverseNested rnn = bucket.getAggregations().get("r" + i);
							((Map<String, Object>) ret.get(key)).put(bucket.getKeyAsString(), rnn.getDocCount());
						}
					}
				}
			}
		}
		return ret;
	}

	@SuppressWarnings("unchecked")
	public Map<String, Object> sumByField(String dsId, String dimension, String metric) throws Exception {
		int level = 0;
		Map<String, Object> ret = new HashMap<>();
		BigDataSource dataSource = bigDataSourceService.get(dsId);
		if (dimension != null && !"".equals(dimension)) {
			FieldMeta target = fieldMetaService.get(dimension);
			FieldMeta metricField = fieldMetaService.get(metric);
			List<String> termLevel = new ArrayList<>();
			AggsBuilder aggs = null;
			if (target.getInNested()) {
				String path = target.getNestedPath();
				AggregationBuilder termAggs = Aggs.terms("2", target.getName())
						.subAggregation(Aggs.sum("sum", metricField.getName()));
				String[] pattern = path.split("[.]");
				termAggs.subAggregation(
						AggregationBuilders.reverseNested("r").subAggregation(Aggs.sum("sumr", metricField.getName())));
				level = pattern.length;
				for (int i = 0; i < level; i++) {
					String rp = pattern[0];
					for (int j = 1; j <= i; j++) {
						rp = rp + "." + pattern[j];
					}
					termLevel.add(rp);
					termAggs.subAggregation(AggregationBuilders.reverseNested("r" + i).path(rp)
							.subAggregation(Aggs.sum("sumr" + i, metricField.getName())));
				}
				aggs = AggsBuilder.build(Aggs.nested("1", path)).subAggs(AggsBuilder.build(termAggs));
			} else {
				aggs = AggsBuilder.build(
						Aggs.terms("1", target.getName()).subAggregation(Aggs.sum("sum", metricField.getName())));
			}
			if (aggs != null) {
				RestHighLevelClient client = esClient.getClient(dataSource.getDwId());
				SearchSourceBuilder aggregation = new SearchSourceBuilder().size(0)
						.aggregation(aggs.getAggregationBuilder());
				System.out.println(aggregation);
				SearchRequest r = new SearchRequest(dataSource.getPath()).source(aggregation);
				SearchResponse response = client.search(r, RequestOptions.DEFAULT);
				Terms t = aggs.getAggsList().getAggsResponse(response);
				List<? extends Bucket> buckets = t.getBuckets();
				Map<String, Object> n = new HashMap<>();
				ret.put("default", n);
				for (Bucket bucket : buckets) {
					Sum v = bucket.getAggregations().get("sum");
					((Map<String, Object>) ret.get("default")).put(bucket.getKeyAsString(), v.getValue());
					if (level != 0) {
						if (!ret.containsKey("root")) {
							Map<String, Object> rm = new HashMap<>();
							ret.put("root", rm);
						}
						ReverseNested rn = bucket.getAggregations().get("r");
						Sum vv = rn.getAggregations().get("sumr");
						((Map<String, Object>) ret.get("root")).put(bucket.getKeyAsString(), vv.getValue());
						for (int i = 0; i < level; i++) {
							String key = termLevel.get(i);
							if (!ret.containsKey(key)) {
								Map<String, Object> rmm = new HashMap<>();
								ret.put(key, rmm);
							}
							ReverseNested rnn = bucket.getAggregations().get("r" + i);
							Sum vvv = rnn.getAggregations().get("sumr" + i);
							((Map<String, Object>) ret.get(key)).put(bucket.getKeyAsString(), vvv.getValue());
						}
					}
				}
			}
		}
		return ret;
	}

	public Map<String, Object> avgByField(String dsId, String dimension, String metric) throws Exception {
		int level = 0;
		Map<String, Object> ret = new HashMap<>();
		BigDataSource dataSource = bigDataSourceService.get(dsId);
		if (dimension != null && !"".equals(dimension)) {
			FieldMeta target = fieldMetaService.get(dimension);
			FieldMeta metricField = fieldMetaService.get(metric);
			List<String> termLevel = new ArrayList<>();
			AggsBuilder aggs = null;
			if (target.getInNested()) {
				String path = target.getNestedPath();
				AggregationBuilder termAggs = Aggs.terms("2", target.getName())
						.subAggregation(Aggs.avg("avg", metricField.getName()));
				String[] pattern = path.split("[.]");
				termAggs.subAggregation(
						AggregationBuilders.reverseNested("r").subAggregation(Aggs.avg("avgr", metricField.getName())));
				level = pattern.length;
				for (int i = 0; i < level; i++) {
					String rp = pattern[0];
					for (int j = 1; j <= i; j++) {
						rp = rp + "." + pattern[j];
					}
					termLevel.add(rp);
					termAggs.subAggregation(AggregationBuilders.reverseNested("r" + i).path(rp)
							.subAggregation(Aggs.avg("avgr" + i, metricField.getName())));
				}
				aggs = AggsBuilder.build(Aggs.nested("1", path)).subAggs(AggsBuilder.build(termAggs));
			} else {
				aggs = AggsBuilder.build(
						Aggs.terms("1", target.getName()).subAggregation(Aggs.avg("avg", metricField.getName())));
			}
			if (aggs != null) {
				RestHighLevelClient client = esClient.getClient(dataSource.getDwId());
				SearchSourceBuilder aggregation = new SearchSourceBuilder().size(0)
						.aggregation(aggs.getAggregationBuilder());
				System.out.println(aggregation);
				SearchRequest r = new SearchRequest(dataSource.getPath()).source(aggregation);
				SearchResponse response = client.search(r, RequestOptions.DEFAULT);
				Terms t = aggs.getAggsList().getAggsResponse(response);
				List<? extends Bucket> buckets = t.getBuckets();
				Map<String, Object> n = new HashMap<>();
				ret.put("default", n);
				for (Bucket bucket : buckets) {
					Avg v = bucket.getAggregations().get("avg");
					((Map<String, Object>) ret.get("default")).put(bucket.getKeyAsString(), v.getValue());
					if (level != 0) {
						if (!ret.containsKey("root")) {
							Map<String, Object> rm = new HashMap<>();
							ret.put("root", rm);
						}
						ReverseNested rn = bucket.getAggregations().get("r");
						Avg vv = rn.getAggregations().get("avgr");
						((Map<String, Object>) ret.get("root")).put(bucket.getKeyAsString(), vv.getValue());
						for (int i = 0; i < level; i++) {
							String key = termLevel.get(i);
							if (!ret.containsKey(key)) {
								Map<String, Object> rmm = new HashMap<>();
								ret.put(key, rmm);
							}
							ReverseNested rnn = bucket.getAggregations().get("r" + i);
							Avg vvv = rnn.getAggregations().get("avgr" + i);
							((Map<String, Object>) ret.get(key)).put(bucket.getKeyAsString(), vvv.getValue());
						}
					}
				}
			}
		}
		return ret;
	}

	public static void main(String[] args) {
		// String path = "n_12.333.212.1.k";
		// String[] pattern = path.split("[.]");
		// String rp = null;
		// for (int i = 0; i < pattern.length; i++) {
		// if (i == 0) {
		// rp = pattern[i];
		// } else {
		// rp = StringUtils.concat(rp, ".", pattern[i]);
		// }
		// System.out.println(rp);
		// }
	}
}
